
package peerToPeerFacebook.proj;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import peerToPeerFacebook.src.edu.washington.cs.cse490h.lib.Node;

/**
 * Node class thats aggregates all functionality. In general this class acts as
 * dispatcher that receives callback coming from Node class and distributes them
 * to other interested classes.
 * 
 */
public final class FBNode extends Node
{
    /**
     * RPC Server thats running on this node. There is only one server that
     * handles all requests from the clients.
     */
    private RPCServer rpcServer = null;

    /** Facebook methods */
    private FacebookMethods fb = null;

    /**
     * For this assignment part we need to make sure that we have two nodes
     * working all the time. It is fine for the node to start, but we need to
     * make sure that this node is restarted at some point. Even after the crash
     * nodes will be recreated as new instances.
     */
    public static double getRecoveryRate()
    {
        return 0 / 100.0;
    }

    /**
     * Here is hard math. One RPC calls writes 3 times to 2 files (master and
     * temp) each time. This means that RPC produces 6 writes. This also means
     * that the file will be opened and written once, this is two places to
     * fail. So for each RPC operation there is 12 places to fail. Accepting
     * friends request takes 6 file operations (2 x get/put pending friends
     * request + 2 updates top friends file). This makes 6 RPC operations or 6 *
     * 12 = 72 failure points. Finally server side put operations will take at
     * least 6 operations. So in total we have that for single request there are
     * 72 + 6 = 78 failure points. So to guarantee at least 50% chance for this
     * to succeed we need to have 78 root 0.5 = 0.9912. Or failure rate should
     * be less than 0.0088
     */
    public static double getFailureRate()
    {
        return 0;
    }

    /** Const that defines paxos group */
    public static final List<Integer> PaxosGroup = Arrays.asList(new Integer[] {1, 2, 3, 4, 5});

    /** Invoker that should be used to invoke calls on paxos group */
    private PaxosClient paxosInvoker = null;

    /**
     * List of rpc clients that are running on this servers. These clients will
     * be notified of any incoming packet
     */
    private Map<Integer, RPCClient> rpcClients = new HashMap<Integer, RPCClient>();

    /**
     * @see peerToPeerFacebook.src.edu.washington.cs.cse490h.lib.Node#start()
     */
    @Override
    public void start()
    {
        // Instantiate server with bound object for file creation.
        // DataManagerCoordinator dmc;
        try
        {
            // Bound RPC server to paxos so that it can be part of paxos group.
            rpcServer = new RPCServer(new NodeWrapper(this), new PaxosProxy(this));
        }
        catch (IOException e)
        {
            throw new InternalError("Failed to load state from the disk", e);
        }
    }

    /**
     * Dispatches incoming packet to all interested parties.
     * 
     * @see peerToPeerFacebook.src.edu.washington.cs.cse490h.lib.Node#onReceive(java.lang.Integer,
     *      int, byte[])
     */
    @Override
    public void onReceive(Integer from, int protocol, byte[] msg)
    {
        // Send the message to everyone that may be interested. It is up to them
        // to
        // figure out if this message is for them or not.
        assert rpcServer != null : "We should not be receiving packets if we do not have server";
        rpcServer.onReceive(from, protocol, msg);

        for (RPCClient client : rpcClients.values())
            client.onReceive(from, protocol, msg);
    }

    /**
     * @see peerToPeerFacebook.src.edu.washington.cs.cse490h.lib.Node#onCommand(java.lang.String)
     */
    @Override
    public void onCommand(String commandString)
    {
        ICommand command = new CommandParser(commandString).parseCommand();
        if (command != null)
        {
            executeCommand(command);
        }
    }

    private void executeCommand(ICommand command)
    {
        // As we are assuming the server is on fixed node 0, this is only for
        // FileServer methods
        RemoteInvoker client = getRPCClient(0);

        // make sure we have interface to call methods on
        if (fb == null)
            fb = new FacebookMethods(this);

        String commandName = command.commandName();

        if (commandName.equalsIgnoreCase(CommandParser.CreateFileCommandName))
        {
            // It should be FileServerCommand
            IFileServerCommand fsCommand = (IFileServerCommand) command;

            IFileServerMethods fsMethods = new FileServerMethods(client, this.addr);
            fsMethods.create(fsCommand.serverID(), fsCommand.fileName());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.GetFileCommandName))
        {
            // It should be FileServerCommand
            IFileServerCommand fsCommand = (IFileServerCommand) command;

            IFileServerMethods fsMethods = new FileServerMethods(client, this.addr);
            fsMethods.get(fsCommand.serverID(), fsCommand.fileName());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.PutFileCommandName))
        {
            // It should be FileServerCommand
            IFileServerCommand fsCommand = (IFileServerCommand) command;

            IFileServerMethods fsMethods = new FileServerMethods(client, this.addr);
            fsMethods.put(fsCommand.serverID(), fsCommand.fileName(), fsCommand.contents());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.AppendFileCommandName))
        {
            // It should be FileServerCommand
            IFileServerCommand fsCommand = (IFileServerCommand) command;

            IFileServerMethods fsMethods = new FileServerMethods(client, this.addr);
            fsMethods.append(fsCommand.serverID(), fsCommand.fileName(), fsCommand.contents());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.DeleteFileCommandName))
        {
            // It should be FileServerCommand
            IFileServerCommand fsCommand = (IFileServerCommand) command;

            IFileServerMethods fsMethods = new FileServerMethods(client, this.addr);
            fsMethods.delete(fsCommand.serverID(), fsCommand.fileName());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.LoginCommandName))
        {
            // It should be FacebookCommand
            IFacebookCommand fbCommand = (IFacebookCommand) command;
            fb.login(fbCommand.userNameOrFriendName());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.LogoutCommandName))
        {
            fb.logout();
        }
        else if (commandName.equalsIgnoreCase(CommandParser.CreateUserCommandName))
        {
            // It should be FacebookCommand
            IFacebookCommand fbCommand = (IFacebookCommand) command;
            fb.createUser(fbCommand.userNameOrFriendName());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.PostMessageToAllCommandName))
        {
            // It should be FacebookCommand
            IFacebookCommand fbCommand = (IFacebookCommand) command;
            fb.postMsgToAllFriends(fbCommand.message());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.SendFriendRequestCommandName))
        {
            // It should be FacebookCommand
            IFacebookCommand fbCommand = (IFacebookCommand) command;
            fb.sendFriendRequest((fbCommand.userNameOrFriendName()));
        }
        else if (commandName.equalsIgnoreCase(CommandParser.AcceptFriendRequestCommandName))
        {
            // It should be FacebookCommand
            IFacebookCommand fbCommand = (IFacebookCommand) command;
            fb.acceptFriendRequest(fbCommand.userNameOrFriendName());
        }
        else if (commandName.equalsIgnoreCase(CommandParser.ReadAllMessagesCommandName))
        {
            fb.readAllMessages();
        }
    }

    /**
     * Returns RPC client that should be used to communicate with the given
     * server. If such client object exists - such object is returned. If it
     * does not exists, new one is created.
     * 
     * It is important to use this method to obtain clients as this will make
     * sure that onReceive messages will be passed to RPCClient. Creating such
     * objects otherwise will not guarantee that.
     * 
     * @param server ServerID for which client should be created
     * @return Client object that should be used to invoke methods.
     */
    public RPCClient getRPCClient(int server)
    {
        if (!rpcClients.containsKey(server))
            rpcClients.put(server, new RPCClient(new NodeWrapper(this), server));

        return rpcClients.get(server);
    }

    /**
     * Retusn paxos invoker that should be used to handle paxos calls
     * 
     * @return Paxos invoker for making calls to paxos group.
     */
    public PaxosClient getPaxosInvoker()
    {
        if (paxosInvoker != null)
            return paxosInvoker;

        RemoteInvoker clients[] = new RPCClient[PaxosGroup.size()];
        ArrayList<Integer> tempArray = new ArrayList<Integer>();
        Random generator = new Random();
       
        while(tempArray.size() != PaxosGroup.size())
        {
         int temp = (generator.nextInt(PaxosGroup.size()) + 1);
         if(!tempArray.contains(temp))
             tempArray.add(temp);
        }  
    
        assert tempArray.size() == PaxosGroup.size() : " The size of the array should now be 3";
        
        for (int i = 0; i < PaxosGroup.size(); i++)
            clients[i] = getRPCClient(tempArray.get(i));
        
        paxosInvoker = new PaxosClient(this, clients);
        return paxosInvoker;
    }
}
